package com.intct.hbase;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.Scanner;


public class SpeedTesting {
    static int limitSpeed =0;

    public static void main(String[] args) throws Exception {
        Scanner scanner = new Scanner(System.in);
        System.out.print("请输入限速（整数）：");
        if (scanner.hasNextInt()) {
            int limitSpeed = scanner.nextInt();
        }else {
            System.out.println("输入的不是整数！");
            System.exit(1);
        }
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Path path = new Path("src/main/resources/data.txt");
        FileSource<String> fs = FileSource.forRecordStreamFormat(new TextLineInputFormat(), path).build();

        DataStreamSource<String> fileStream = env.fromSource(fs, WatermarkStrategy.noWatermarks(), "data");

        SingleOutputStreamOperator<Map<String, Object>> mapStream = fileStream.map(new MapFunction<String, Map<String, Object>>() {
            @Override
            public Map<String, Object> map(String line) throws Exception {
                Map<String, Object> map = JSON.parseObject(line);
                // 将字符串时间转为时间戳（毫秒）
                String timeStr = (String) map.get("time");
                LocalDateTime dateTime = LocalDateTime.parse(
                        timeStr,
                        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
                );
                long timestamp = dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                map.put("timestamp", timestamp); // 将时间戳存入 Map
                return map;
            }
        });

        SingleOutputStreamOperator<Map<String, Object>> waterStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<Map<String, Object>>forBoundedOutOfOrderness(Duration.ofSeconds(300))
                .withTimestampAssigner((m, l) -> (long) m.get("timestamp")));


        OutputTag<Tuple3<Map<String, Object>, Integer, Integer>> safeSpeed =
                new OutputTag<Tuple3<Map<String, Object>, Integer, Integer>>("未超速记录") {};

        SingleOutputStreamOperator<Tuple3<Map<String, Object>, Integer, Integer>> warnSpeed =
                waterStream.keyBy(map -> map.get("car"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<Map<String, Object>, Tuple3<Map<String, Object>, Integer, Integer>, Object, TimeWindow>() {

                    @Override
                    public void process(Object o, ProcessWindowFunction<Map<String, Object>,
                            Tuple3<Map<String, Object>, Integer, Integer>, Object, TimeWindow>.Context context, Iterable<Map<String, Object>> iterable, Collector<Tuple3<Map<String, Object>, Integer, Integer>> collector) throws Exception {
                            int a = 0;
                            int b = 0;
                            Map<String, Object> resultMap = null;

                            for (Map<String, Object> event : iterable) {
                                resultMap = event;
                                long speed = Long.parseLong(event.get("speed").toString());
                                if (speed > limitSpeed) {
                                    a++;
                                } else {
                                    b++;
                                }
                            }

                            Tuple3<Map<String, Object>, Integer, Integer> result = Tuple3.of(resultMap, a, b);

                            if (a >= 1) collector.collect(result);
                            if (b >= 1) context.output(safeSpeed, result);
                        }

                });

        SingleOutputStreamOperator<Tuple3<Map<String, Object>, Integer, Integer>> finalStream_over =
                warnSpeed.keyBy(t -> t.f0.get("car"))
                .reduce(new ReduceFunction<Tuple3<Map<String, Object>, Integer, Integer>>() {
                    @Override
                    public Tuple3<Map<String, Object>, Integer, Integer> reduce(Tuple3<Map<String, Object>, Integer, Integer> t1, Tuple3<Map<String, Object>, Integer, Integer> t2) throws Exception {
                        return Tuple3.of(t1.f0, t1.f1 + t2.f1, t2.f1 + t2.f2);
                    }
                });
        SingleOutputStreamOperator<Tuple3<Map<String, Object>, Integer, Integer>> finalStream_safe =
                warnSpeed.getSideOutput(safeSpeed).keyBy(t -> t.f0.get("car"))
                .reduce(new ReduceFunction<Tuple3<Map<String, Object>, Integer, Integer>>() {
                    @Override
                    public Tuple3<Map<String, Object>, Integer, Integer> reduce(Tuple3<Map<String, Object>, Integer, Integer> t1, Tuple3<Map<String, Object>, Integer, Integer> t2) throws Exception {
                        return Tuple3.of(t1.f0, t1.f1 + t2.f1, t2.f1 + t2.f2);
                    }
                });

        finalStream_over.print("超速记录：");
        finalStream_safe.print("未超速记录：");

        env.execute();
        env.close();
    }
}
